package cn.monkey.io.transport.core;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.function.Function;
import java.util.function.Predicate;

public class AbstractDelegatesExchange<T extends Exchange>
        implements DelegatesExchange<T> {

    private final String id;
    private final List<T> delegates;
    private final ExchangeContext exchangeContext;

    public AbstractDelegatesExchange(String id, List<T> delegates) {
        this.id = id;
        this.delegates = Collections.unmodifiableList(delegates);
        this.exchangeContext = delegates
                .stream()
                .map(Exchange::context)
                .filter(Objects::nonNull)
                .reduce(CompositeExchangeContext::new)
                .orElseThrow(() -> new IllegalArgumentException("delegates context is empty"));
    }

    @Override
    public List<T> delegates() {
        return this.delegates;
    }

    @Override
    public String id() {
        return this.id;
    }

    @Override
    public <O> Flux<O> receive(Class<O> clazz,
                               Predicate<O> p) {
        return Flux.fromStream(this.delegates()
                        .stream().map(d -> d.receive(clazz, p)))
                .flatMap(Function.identity());
    }

    @Override
    public Mono<Void> write(Object data) {
        List<Mono<Void>> list = this.delegates()
                .stream()
                .map(d -> d.write(data))
                .toList();
        return Mono.when(list);
    }

    @Override
    public Mono<Void> writeAndFlush(Object data) {
        List<Mono<Void>> list = this.delegates()
                .stream()
                .map(d -> d.writeAndFlush(data))
                .toList();
        return Mono.when(list);
    }

    @Override
    public Mono<Void> flush() {
        List<Mono<Void>> list = this.delegates()
                .stream()
                .map(Exchange::flush)
                .toList();
        return Mono.when(list);
    }

    @Override
    public ExchangeContext context() {
        return this.exchangeContext;
    }

    @Override
    public Inbound inbound() {
        throw new UnsupportedOperationException();
    }

    @Override
    public Outbound outbound() {
        throw new UnsupportedOperationException();
    }

    @Override
    public Mono<Void> close() {
        List<Mono<Void>> list = this.delegates()
                .stream()
                .map(Exchange::close)
                .toList();
        return Mono.when(list);
    }

    @Override
    public boolean isActive() {
        return !this.delegates().isEmpty() || this.delegates().stream()
                .anyMatch(Exchange::isActive);
    }

    @Override
    public DelegatesExchange<T> bind() {
        this.context().setAttribute(Exchange.KEY, this);
        return this;
    }
}
